package com.intct.flink.study;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author gufg
 * @since 2025-10-14 14:08
 */
public class SQLTest12 {
    public static void main(String[] args) {
        // 配置属性
        Configuration conf = new Configuration();
        conf.set(RestOptions.BIND_PORT, "8081");

        // 1 获取取环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        // 并行度
        env.setParallelism(1);
        // 开启检点
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        // 获取SQL执行环境
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 映射mysql的订单
        tenv.executeSql(
                "create table flink_test01( " +
                        // 如果是Flink sql的关键字，需要用反引号转义
                        "`user`              INT " +
                        ",order_time    TIMESTAMP(3) " +
                        ",proctime as procTime()           " +
                        ",watermark for order_time as order_time - interval '0' second " +
                        ",PRIMARY KEY (`user`) NOT ENFORCED " +
                        ") with ( " +
                        "  'connector' = 'mysql-cdc' " +
                        "  ,'hostname' = 'cdh-node' " +
                        "  ,'port' = '13306' " +
                        "  ,'username' = 'root' " +
                        "  ,'password' = 'Test_090110' " +
                        "  ,'database-name' = 'm1' " +
                        "  ,'table-name' = 'orders_info' " +
                        ") ");


        tenv.executeSql("create table redis_table " +
                " (                        " +
                "     name STRING          " +
                "     ,score DOUBLE        " +
                " 	,v11 STRING          " +
                " 	                       " +
                " )                        " +
                " with                     " +
                " (                        " +
                "     'connector'='redis'  " +
                " 	,'host'='cdh-node' " +
                " 	,'port'='16379'         " +
                " 	,'password'='Test_090110'  " +
                " 	,'redis-mode'='single' " +
                " 	,'command'='zadd'      " +
                " 	,'maxIdle'='2'      " +
                " 	,'minIdle'='1'      " +
//                " 	,'lookup.cache.max-rows'='10'      " +
                " 	,'lookup.cache.ttl'='10'      " +
                " 	,'max.retries'='3'      " +
                " )                        "
        );
        
        tenv.executeSql("create table sink_table(" +
                "    username varchar" +
                "    , level varchar" +
                ") with (" +
                "    'connector'='print'" +
                ")");

        tenv.executeSql(" insert into sink_table SELECT                        " +
                        "    cast(s.`user` as string)                                   " +
                        " 	,d.name                                                     " +
                        " FROM                                                          " +
                        " flink_test01 s                                                " +
                        " LEFT JOIN redis_table FOR SYSTEM_TIME AS OF s.proctime as d   " +
                        " ON CAST(`user` AS STRING) = d.v11                            "
        );

//        tenv.executeSql("select * from sink_table").print();
    }
}
